package com.atguigu.flink.function;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 每个处理函数都有一个对应的类， rich + 接口名 ， 例如MapFunction ==> RichMapFunction
 *  1.富含数带有生命周期
 *  open  ： 在生命周期开始的时候执行的方法，一般用于做资源的初始化 /每一个并行度（并行子任务）上执行一次
 *  close ： 在生命周期结束的时候执行的方法，一般用于做资源的关闭（无界数据流永远不会执行）
 *  map：每来一次数据执行一次
 *
 */

public class UdfRichFunction {

    public static void main(String[] args) throws Exception {
        // TODO 1. 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        // TODO 2. 从指定网络端口读取数据
        DataStreamSource<String> socketTextStream = env.socketTextStream("hadoop102", 8888);

        // TODO 3. 对数据进行处理
        // TODO 4. 打印输出
        /*socketTextStream
                .map(
                new MapFunction<String, String >() {
                    @Override
                    public String map(String value) throws Exception {
                        return "atguigu ->" + value;
                    }
                }
        ).print();*/


        socketTextStream.map(
                new RichMapFunction<String, String >() {

                    //生命周期
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        System.out.println("----open生命周期开始----");
                        RuntimeContext runtimeContext = getRuntimeContext();
                        System.out.println(runtimeContext.getIndexOfThisSubtask());
                        System.out.println(runtimeContext.getTaskNameWithSubtasks());
                    }

                    @Override
                    public String map(String value) throws Exception {
                        return "atguigu ->" + value;
                    }

                    @Override
                    public void close() throws Exception {
                        System.out.println("----close生命周期结束----");
                    }
                }
        );

        // TODO 5. 提交作业
        env.execute();
    }
}
